
/****************************************************************************
**
** Copyright (C) 2004-2008 VATATA.com . All rights reserved.
**
** This file is the part of the Miniutil Poject of VATATA.com.
** Vatata's miniutil library is lightweight and portable (for linux
** and Windows). The C++ Class Library included Threadpool, Dll danymic loading, 
** General socket application model and tcp server, httpserver/servlet, etc. 
** It is a pratical utility disigned for who don't want learn the Big C++ 
** Developing Platform, such as  Boost or ACE.
**
** This file may be used under the terms of the GNU General Public
** License versions 2.0 or 3.0 as published by the Free Software
** Foundation and appearing in the files LICENSE.GPL2 and LICENSE.GPL3
** included in the packaging of this file.  Alternatively you may (at
** your option) use any later version of the GNU General Public
** License if such license has been publicly approved by VATATA.com.
** (or its successors, if any) and the Miniutil Project Aministrators. 
**
** Please review the following information to ensure GNU General
** Public Licensing requirements will be met:
** http://www.gnu.org/licenses/lgpl.html. If you are unsure which 
** license is appropriate for your use, please review the following 
** information: http://code.google.com/p/miniutil/ or contact 
** http://www.vatata.com.
**
** In addition, as a special exception, VATATA.com, as the sole
** copyright holder for Miniutil Project, grants users of VATATA
** P2P Platform the right to release and package with the related 
** libraries without the source code.
**
** This file is provided "AS IS" with NO WARRANTY OF ANY KIND,
** INCLUDING THE WARRANTIES OF DESIGN, MERCHANTABILITY AND FITNESS FOR
** A PARTICULAR PURPOSE. vatata reserves all rights not expressly
** granted herein.
**
** This file is provided AS IS with NO WARRANTY OF ANY KIND, INCLUDING THE
** WARRANTY OF DESIGN, MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE.
**
****************************************************************************/


/*

*/

#include "miniutil/task.h"
#include "miniutil/date.h"
#include "miniutil/log.h"

#include "miniutil/timer.h"

using namespace miniutil::sys;

task_error::task_error( const std::string & msg) : std::runtime_error(msg)
{
	M_ERROR("miniutil::sys::task_manager error: %s\n", msg.c_str() );
}


task_manager::task_manager()// : running( true ), manager(), workers()
{
	running = true;
	manager.start( *this );
}

task_manager::~task_manager()
{
	running = false;
	this->task_queue.clear();

	M_DEBUG("[ ~task_manager ] .");
	while ( manager.get_state() == miniutil::thread::STATE_RUNNING )
	{
		M_DEBUG(" . ");
		miniutil::auto_lock tm( task_mutex );
		tm.notifyAll();
		tm.wait( 1000 );
	}
	M_DEBUG(" stopped!\n");
}

void task_manager::operator()()
{
	miniutil::auto_lock tm( task_mutex );
	while( running )
	{
		if ( task_queue.size() == 0 )
		{
			tm.wait();
			continue;
		}
		miniutil::m_uint32_t cur_ms = miniutil::date::uptime_ms();

		_task_type * ptask = task_queue.front();
		if ( cur_ms < ptask->execute_ms )
		{
			tm.wait( ptask->execute_ms - cur_ms );
			continue;
		}

		//M_DEBUG("miniutil::sys::task_manager run %p ",  ptask->p_task );
		if ( ! workers.runTask( ptask->p_task ) )
		{
			M_DEBUG("miniutil::sys::task_manager thread pool is busy! wait and retry !\n");
			tm.wait( RETRY_INTERVAL );
			continue;
		}
		//M_DEBUG(" ok\n");

		// everything is ok, the task_manager is running in a thread!
		task_queue.pop_front();
		if ( ptask->interval_ms == 0 || ptask->loop_time == 1 )
		{
			// not loop task_manager or loop task_manager shoulb be over!
			delete ptask;
			continue;
		}

		// re-insert loop task_manager.
		if ( ptask->loop_time > 1 ) ptask->loop_time --;
		ptask->execute_ms = cur_ms + ptask->interval_ms;

		std::list<_task_type *>::iterator it = this->task_queue.begin();
		while( it != this->task_queue.end() )
		{
			if ( ptask->execute_ms < (*it)->execute_ms )
				break;
			it++;
		}
		this->task_queue.insert( it, ptask );
	}
	tm.notifyAll();
}

size_t task_manager::count()
{
	return this->task_queue.size();
}
void task_manager::clear()
{
	miniutil::auto_lock tm( task_mutex );
	this->task_queue.clear();
}

void task_manager::call( miniutil::runnable *p_task )
{
	if ( ! workers.runTask( p_task ) )
	{
		delay( p_task );
	}
}

void task_manager::delay( miniutil::runnable * p_task, unsigned int delay_ms )
{
	miniutil::m_uint32_t cur_ms = miniutil::date::uptime_ms();

	_task_type *ptask = new _task_type();
	ptask->p_task = p_task;
	ptask->execute_ms = cur_ms + delay_ms;
	ptask->interval_ms = 0;
	ptask->loop_time = 1;

	miniutil::auto_lock tm( task_mutex );
	std::list<_task_type *>::iterator it = this->task_queue.begin();
	while( it != this->task_queue.end() )
	{
		if ( ptask->execute_ms < (*it)->execute_ms )
			break;
		it++;
	}
	this->task_queue.insert( it, ptask );
	tm.notifyAll();
}

void task_manager::loop( miniutil::runnable * p_task,
				miniutil::m_uint32_t delay_ms , 
				miniutil::m_uint32_t interval_ms , 
				miniutil::m_uint32_t loop_time )
{
	miniutil::m_uint32_t cur_ms = miniutil::date::uptime_ms();

	_task_type *ptask = new _task_type();
	ptask->p_task = p_task;
	ptask->execute_ms = cur_ms + delay_ms;
	ptask->interval_ms = interval_ms ;
	ptask->loop_time = loop_time;

	miniutil::auto_lock tm( task_mutex );
	std::list<_task_type *>::iterator it = this->task_queue.begin();
	while( it != this->task_queue.end() )
	{
		if ( ptask->execute_ms < (*it)->execute_ms )
			break;
		it++;
	}
	this->task_queue.insert( it, ptask );
	tm.notifyAll();
}

int  task_manager::check( miniutil::runnable *p_task )
{
	miniutil::auto_lock tm( task_mutex );
	std::list<_task_type *>::iterator it = this->task_queue.begin();
	while( it != this->task_queue.end() )
	{
		if ( (*it)->p_task == p_task )
		{
			if ( (*it)->interval_ms == 0 )
				return TYPE_DELAY;
			return TYPE_LOOP;
		}
		it++;
	}
	return TYPE_NONE;
}

bool task_manager::cancel( miniutil::runnable *p_task )
{

    int cancelcount = 0;
    {
	    miniutil::auto_lock tm( task_mutex );
	    std::list<_task_type *>::iterator it = this->task_queue.begin();
	    while( it != this->task_queue.end() )
	    {
		    _task_type * ptask = *it;
		    if ( ptask->p_task == p_task )
		    {
			    this->task_queue.erase( it ++ );
                cancelcount ++;
		    }
            else
    		    it++;
	    }
    }

    //while( p_task->get_state() == miniutil::runnable::STATE_RUNNING )
    //{
    //    M_WARN("cancel task when task is running, wait task is finished...\n" );
    //    thread::sleep( 400 );
    //}

    if ( cancelcount == 0 )
	    return false;
    return true;
}

void task_manager::dump()
{
	miniutil::auto_lock tm( task_mutex );
	std::list<_task_type *>::iterator it = this->task_queue.begin();
	M_DEBUG( "\n[miniutil::sys::task_manager::dump()]\n");
	while( it != this->task_queue.end() )
	{
		M_DEBUG( "TASK[%p] exectime [%d] interval [%d] loop [%d]\n", 
			(*it)->p_task,
			(*it)->execute_ms,
			(*it)->interval_ms,
			(*it)->loop_time );
		it++;
	}
	M_DEBUG("\n");
}

static task_manager * _sys_timer = NULL;

void miniutil::sys::setTimer( miniutil::runnable * t, int interval_ms )
{
    if ( _sys_timer == NULL ) 
        _sys_timer = new task_manager();

    _sys_timer->loop( t, interval_ms, interval_ms );
}

void miniutil::sys::setTimeout( miniutil::runnable * t ,  int timeout_ms)
{
    if ( _sys_timer == NULL ) 
        _sys_timer = new task_manager();

    _sys_timer->delay( t, timeout_ms );
}

void miniutil::sys::clearTimer( miniutil::runnable * t )
{
    if ( _sys_timer == NULL ) 
        return;
    _sys_timer->cancel( t );
}
